package com.atguigu.gmall.realtime.app.dwd;

import com.atguigu.gmall.realtime.app.BaseSQLApp;
import com.atguigu.gmall.realtime.common.Constant;
import com.atguigu.gmall.realtime.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

/**
 * @Author lzc
 * @Date 2022/12/7 10:17
 */
    public class Dwd_04_DwdTradeOrderCancel extends BaseSQLApp {
        public static void main(String[] args) {
            new Dwd_04_DwdTradeOrderCancel().init(
                3004,
                2,
                "Dwd_04_DwdTradeOrderCancel"
            );
        }
        
        
        @Override
        protected void handle(StreamExecutionEnvironment env,
                              StreamTableEnvironment tEnv) {
            tEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(30 * 60 + 5));
            
            // 1. 读取 ods_db (过滤出来 order_info 中的订单取消记录)
            readOdsDb(tEnv, "dwd_trade_order_detail");
            
            // 2. 读取 dwd_trade_order_detail, 可以获取到详情, 详情活动,详情优惠券,字典表已经 join 好的数据
            // 使用的普通的 kafka, 所以读取的 left join 的结果会有重复数据, 将来在 dws 层如果业务需要,则考虑去重
            tEnv.executeSql("create table dwd_trade_order_detail(" +
                                "id string," +
                                "order_id string," +
                                "user_id string," +
                                "sku_id string," +
                                "sku_name string," +
                                "province_id string," +
                                "activity_id string," +
                                "activity_rule_id string," +
                                "coupon_id string," +
                                "date_id string," +
                                "create_time string," +
                                "source_id string," +
                                "source_type string," +
                                "source_type_name string," +
                                "sku_num string," +
                                "split_original_amount string," +
                                "split_activity_amount string," +
                                "split_coupon_amount string," +
                                "split_total_amount string," +
                                "ts bigint " +
                                ")" + SQLUtil.getKafkaSourceDDL(Constant.TOPIC_DWD_TRADE_ORDER_DETAIL, "Dwd_04_DwdTradeOrderCancel"));
            
            //        tEnv.sqlQuery("select * from dwd_trade_order_detail").execute().print();
            
            // 3. 过滤订单取消记录
            Table orderCancelInfo = tEnv.sqlQuery("select " +
                                                      "data['id'] id," +
                                                      "data['user_id'] user_id," +
                                                      "data['province_id'] province_id," +
                                                      "data['operate_time'] operate_time," +
                                                      "ts " +
                                                      "from ods_db " +
                                                      "where `database`='gmall2022' " +
                                                      "and `table`='order_info' " +
                                                      "and `type`='update' " +
                                                      "and `old`['order_status']='1001' " +
                                                      "and `data`['order_status']='1003'");
            
            tEnv.createTemporaryView("order_cancel_info", orderCancelInfo);
            // 4. 两张表 join   order_cancel_info 内连接 dwd_trade_order_detail
            Table result = tEnv.sqlQuery("select " +
                                             "od.id," +
                                             "od.order_id," +
                                             "oci.user_id," +
                                             "od.sku_id," +
                                             "od.sku_name," +
                                             "oci.province_id," +
                                             "od.activity_id," +
                                             "od.activity_rule_id," +
                                             "od.coupon_id," +
                                             "date_format(oci.operate_time, 'yyyy-MM-dd') operate_date_id," +
                                             "oci.operate_time," +
                                             "od.source_id," +
                                             "od.source_type," +
                                             "od.source_type_name," +
                                             "od.sku_num," +
                                             "od.split_original_amount," +
                                             "od.split_activity_amount," +
                                             "od.split_coupon_amount," +
                                             "od.split_total_amount," +
                                             "oci.ts " +  // 订单取消的时候的时间戳
                                             "from order_cancel_info oci " +
                                             "join dwd_trade_order_detail od " +
                                             "on oci.id=od.order_id");
            
            // 5. 写出到kafka 中
            tEnv.executeSql("create table dwd_trade_cancel_detail(" +
                                    "id string," +
                                    "order_id string," +
                                    "user_id string," +
                                    "sku_id string," +
                                    "sku_name string," +
                                    "province_id string," +
                                    "activity_id string," +
                                    "activity_rule_id string," +
                                    "coupon_id string," +
                                    "date_id string," +
                                    "cancel_time string," +
                                    "source_id string," +
                                    "source_type_code string," +
                                    "source_type_name string," +
                                    "sku_num string," +
                                    "split_original_amount string," +
                                    "split_activity_amount string," +
                                    "split_coupon_amount string," +
                                    "split_total_amount string," +
                                    "ts bigint " +
                                    ")" + SQLUtil.getKafkaSinkDDL(Constant.TOPIC_DWD_TRADE_CANCEL_DETAIL));
            result.executeInsert("dwd_trade_cancel_detail");
        }
    }
/*
交易域取消订单事务事实表
order_detail 不会变化
order_info  update
order_detail_activity 不会变化
order_detail_coupon 不会变化

两种方式:
    1. 类似下单事务事实表
        5 张表 join  (order_info 去的 update并 order_status变化 1001=>1003)
        
    2. 使用 order_info  去 join 下单事务事实表(上一个需求的 结果)

 */